package com.gt.ln

import com.gt.SCUtil
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD

//页面转化率
object Spark_hot_product_05 {

  //用户访问动作表
  case class UserVisitAction(
                              date: String, //用户点击行为的日期
                              user_id: Long, //用户的 ID
                              session_id: String, //Session 的 ID
                              page_id: Long, //某个页面的 ID
                              action_time: String, //动作的时间点
                              search_keyword: String, //用户搜索的关键词
                              click_category_id: Long, //某一个商品品类的 ID
                              click_product_id: Long, //某一个商品的 ID
                              order_category_ids: String, //一次订单中所有品类的 ID 集合
                              order_product_ids: String, //一次订单中所有商品的 ID 集合
                              pay_category_ids: String, //一次支付中所有品类的 ID 集合
                              pay_product_ids: String, //一次支付中所有商品的 ID 集合
                              city_id: Long
  )

  def main(args: Array[String]): Unit = {
    val sc: SparkContext = SCUtil.createLocalSc()

    //1. 获取数据
    val rdd: RDD[String] = sc.textFile("data/user_visit_action.txt")

    //2. 数据格式转换
    val srcRdd: RDD[UserVisitAction] = rdd.map(line => {
      val arr: Array[String] = line.split("_")
      UserVisitAction(arr(0), arr(1).toLong, arr(2), arr(3).toLong, arr(4),
        arr(5), arr(6).toLong, arr(7).toLong, arr(8), arr(9)
        , arr(10), arr(11), arr(12).toLong)
    })

    //TODO 只需要指定跳转的转换率
    val ids = List(1,2,3,4,5,6,7)
    val idFlows = ids.zip(ids.tail) //List((1,2),(2,3),(3,4),(4,5),(5,6),(6,7))

    //3. 统计所有页面的访问次数 (page1, 100)
    val pageIdCount: RDD[(Long, Long)] = srcRdd.filter(data => ids.init.contains(data.page_id))
      .map(actionBean => {
      (actionBean.page_id, 1l)
    }).reduceByKey(_ + _)

    //(pageId, pageIdCount)
    val pageIdCountMap: Map[Long, Long] = pageIdCount.collect().toMap

    //4. 统计用户单次访问中页面连续跳转数据  ((pageId1, pageId2),count)
    val rdd2: RDD[((Long, Long), Long)] = srcRdd.groupBy(_.session_id)
      .map(data => {
        val pageIdList: List[Long] = data._2.toList.sortBy(_.action_time).map(_.page_id)
        pageIdList.zip(pageIdList.tail)
      }).flatMap(x => x)
      .filter(data => {idFlows.contains(data)})
      .map((_,1l))
      .reduceByKey(_+_)

    val rddResult: RDD[String] = rdd2.map(data => {
      val pageId1 = data._1._1
      val pageId2 = data._1._2
      val count = data._2
      //从页面1 跳转到 页面2 的概率
      val persent = count.toDouble / pageIdCountMap.get(pageId1).get
      s"${pageId1} --> ${pageId2} 转化率为: ${persent} "
    })

    rddResult.collect().foreach(println)




    sc.stop()
  }

}
